/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.UnknownHostException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import java.util.zip.CRC32;
import java.util.zip.CheckedInputStream;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.pipe.BufferPool;
import com.nhncorp.neptune.changelog.pipe.ChangeLogFileChannel;
import com.nhncorp.neptune.changelog.pipe.Pipe;
import com.nhncorp.neptune.changelog.pipe.PipeEventHandler;
import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.common.ipc.NRPC.Server;
import com.nhncorp.neptune.common.testhelper.FaultInjectionProxy;
import com.nhncorp.neptune.common.testhelper.FaultManager;
import com.nhncorp.neptune.common.util.NetworkUtil;
import com.nhncorp.pleiades.client.LockEventData;
import com.nhncorp.pleiades.client.LockEventFilter;
import com.nhncorp.pleiades.client.LockEventMonitor;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockService.LockMode;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class ChangeLogServer implements ChangeLogServerIF {
  public static boolean IS_TEST_MODE = false;
  public static final int TIMEOUT = 10000; // 10 sec
  static final Log LOG = LogFactory.getLog(ChangeLogServer.class);
  static final int numProcessors = Runtime.getRuntime().availableProcessors();
  public static final int PORT_DIFF = 10;
  static FaultManager fm;

  final Listener listener;
  final Server server;
  final Map<String, HashMap<String, Pipe>> tabletPipeMap = new HashMap<String, HashMap<String, Pipe>>();
  final String changeLogStorePath;
  final NConfiguration conf;

  String lockPath;
  LockService lockService;
  int rpcPort;

  public ChangeLogServer(NConfiguration conf) throws IOException {
    this.conf = conf;
    rpcPort = conf.getInt("neptune.changelog.server.port", 57001);
    int handlerCount = conf.getInt("changeLogServer.handler.count", numProcessors * 2);
    changeLogStorePath = getLogStorePath(conf);

    listener = new Listener(conf, changeLogStorePath);
    server = NRPC.getServer(lockService, this, "0.0.0.0", rpcPort, handlerCount, false, conf);
  }

  private String getHostInfo() throws UnknownHostException {
    try {
      String hostName = InetAddress.getLocalHost().getHostName();
      InetAddress.getByName(hostName);
      return hostName;
    } catch (UnknownHostException e) {
      return InetAddress.getLocalHost().getHostAddress();
    }
  }

  private String getLogStorePath(NConfiguration conf) throws IOException {
    String path = conf.get("changelog.data.dir"
        , System.getProperty("user.home") + File.separator + ".changelog");

    if (!path.endsWith("/")) {
      path = path + "/";
    }

    String[] pathElements;
    if ((pathElements = path.split("\\$USER_HOME\\$")).length == 1) {
      path = pathElements[0];
    } else if (pathElements.length == 2) {
      path = pathElements[0] + System.getProperty("user.home") + pathElements[1];
    }

    File dir = new File(path);

    if (dir.exists() && dir.canWrite() == false) {
      throw new IOException("No permission to write change logs in path [" + path + "]");
    }

    return path;
  }
  
  private void deleteLock() {
    try {
      if (lockService.deleteNode(lockPath, true) == false) {
        LOG.error("fail to delete lock path [" + lockPath + "]");
      }
    } catch (IOException e) {
      LOG.error("fail to delete node lockpaht [" + lockPath + "]");
    }
  }

  private void initLockService(NConfiguration conf, String hostName) throws IOException {
    lockPath = Constants.CHANGELOG_SERVER + "/" + NetworkUtil.hostNameToFileName(hostName);
    String owner = "ChangeLogServer:" + hostName;

    LOG.info("Create node of Pleiades lock at [" + lockPath + "], service type ["
        + conf.get("pleiades.service.name", "neptune") + "], lock owner [" + owner + "]");

    lockService = LockServiceFactory.getLockService(conf, owner, true);
    addLockEventHandler(conf, owner);

    int retry = 0;

    while (true) {
      if (lockService.getLock(lockPath, LockMode.LOCK_EXCLUSIVE)) {
        LockMetaData lockMetaData = new LockMetaData(lockPath);
        lockMetaData.setLockMode(LockService.LockMode.LOCK_EXCLUSIVE.ordinal());

        if (lockService.createNode(lockMetaData, true)) {
          LOG.info("Succeed to register change log server info to lockPath [" + lockPath + "]");
          break;
        }
      }

      if (retry >= 5) {
        LOG.fatal("Can't get server lock:" + lockPath + ". ChangeLogServer killed");
        throw new IOException("Can't get server lock:" + lockPath + ". ChangeLogServer killed");
      }

      LockMetaData metaData = lockService.getMetaData(lockPath);

      LOG.warn("Can't get server lock:" + lockPath + ". Lock Creator : "
          + (metaData == null ? "null" : metaData.getCreator()) + ", but retry");

      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        LOG.warn(e);
      }
      retry++;
    }
    
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      LOG.warn(e);
    }
  }

  private void addLockEventHandler(final NConfiguration conf, final String owner)
      throws IOException {
    LockEventMonitor showdownMonitor = new LockEventMonitor() {
      public LockEventFilter getFilter() {
        LockEventFilter filter = new LockEventFilter();
        filter.addConstraint(LockEventFilter.Operator.PF, lockPath);

        List<LockService.Events> lockEvents = new ArrayList<LockService.Events>();
        lockEvents.add(LockService.Events.SESSION_EXPIRED);
        lockEvents.add(LockService.Events.NODE_REMOVED);
        filter.setLockEventList(lockEvents);

        return filter;
      }

      public void notifyEvent(LockService.Events eventType, LockEventData eventData) {
        try {
          checkForShutdown(conf, owner);
        } catch (Exception e) {
          LOG.error(e);
        }
      }
    };

    if (LockServiceFactory.getLockService(conf, owner, true).getEventHandler() != null) {
      LockServiceFactory.getLockService(conf, owner, true).getEventHandler()
          .addLockEventMonitor(showdownMonitor);
    }
  }

  private void checkForShutdown(NConfiguration conf, String hostName) throws IOException {
    // Session이 expire되는 경우 네트워크 문제 등으로 인해 Pleiades에 lock을 유지 못하는 경우이다.
    // 이 경우 Server는 lock 파일이 존재하면 lock을 획득하는 시도를 하고
    // lock 파일이 없는 경우 자동으로 종료한다.
    while (true) {
      // lock 파일이 존재하지 않으면 종료
      LockService lockService = LockServiceFactory.getLockService(conf, hostName, true);
      if (lockService.getMetaData(lockPath) == null) {
        LOG.fatal("ChangeLogServer stopped. session expired:" + lockPath);
        stop();
      }

      // lock을 다시 획득할 수 있으면 정상적으로 계속 수행
      if (lockService.getLock(lockPath, LockService.LockMode.LOCK_EXCLUSIVE)) {
        return;
      }
      try {
        Thread.sleep(1 * 1000);
      } catch (InterruptedException e) {
        return;
      }
    }
  }

  public void start() throws IOException {
    server.start();
    listener.start();
    try {
      initLockService(conf, getHostInfo() + ":" + rpcPort);
    } catch(IOException e) {
      this.stop();
      throw e;
    }
  }

  public void stop() {
    BufferPool.singleton().stop();

    synchronized (tabletPipeMap) {
      for (String tabletName : tabletPipeMap.keySet()) {
        closeAll(tabletName);
      }

      try {
        Thread.sleep(1000);
      } catch (Exception e) {
      }

      tabletPipeMap.clear();
    }
    listener.stopListening();

    try {
      //deleteLock();
      lockService.close();
    } catch (Exception e) {
      LOG.debug("exception in closing lock service", e);
    }
    server.stop();
  }

  public int getPort() {
    return listener.getPipePort();
  }

  private static void formatAll() throws IOException {
    LOG.info("format all the meta info of change log servers");
    long startTime = System.currentTimeMillis();

    NConfiguration conf = new NConfiguration();
    // 이전의 모든 lock 정보를 삭제할때 까지 반복 수행
    while (true) {
      try {
        Thread.sleep(5 * 1000);
      } catch (InterruptedException e1) {
        return;
      }

      LockService lockService = null;
      try {
        lockService = LockServiceFactory.getLockService(conf, "NeptuneMaster", false);

        if (lockService != null) {
          if (!lockService.exist(Constants.CHANGELOG_SERVER)) {
            // 아무것도 없는 경우
            LOG.debug("no lock node");
            return;
          } else {
            boolean lockResult = lockService.deleteNode(Constants.CHANGELOG_SERVER, true);
            if (lockResult) {
              LOG.debug("deleted CHANGELOG_SERVER files and lock");
              lockService.releaseAll();
              return;
            }
          }
        } else {
          LockServiceFactory.clear("NeptuneMaster");
        }

        if (System.currentTimeMillis() - startTime > 30 * 1000) {
          LOG.error("Can't format. timeout");
          return;
        }
      } catch (Exception e) {
        LOG.error(e.getMessage(), e);
      } finally {
        if (lockService != null) {
          LockServiceFactory.close("NeptuneMaster");
        } else {
          LOG.fatal("Can't get LockServer: lock opwner=NeptuneMaster");
        }
      }
    }
  }

  public static void main(String[] args) {
    if (args.length > 0 && "-format".equals(args[0])) {
      try {
        formatAll();
      } catch (IOException e) {
        e.printStackTrace();
      }
      System.exit(0);
    }

    ChangeLogServerIF server;

    NConfiguration conf = new NConfiguration();

    if (args.length > 0) {
      conf.set("neptune.changelog.server.port", conf.getInt("neptune.changelog.server.port", 57001)
          + Integer.valueOf(args[0]));
      if ((args.length > 1 && args[1].equalsIgnoreCase("true")) 
          || conf.get("testmode").equalsIgnoreCase("true")) {
        ChangeLogServer.IS_TEST_MODE = true;
      }
    }

    try {
      server = ChangeLogServer.create(conf);
      server.start();
      LOG.info("Change log server is ready... ");
    } catch (IOException e) {
      LOG.error("Initializing change log server is fail", e);
      System.exit(-1);
    }
  }

  private static ChangeLogServerIF create(NConfiguration conf) throws IOException {
    ChangeLogServerIF server;

    if (ChangeLogServer.IS_TEST_MODE) {
      fm = FaultManager.create(conf);
      server = FaultInjectionProxy.wrap(new ChangeLogServer(conf), ChangeLogServerIF.class);
    } else {
      server = new ChangeLogServer(conf);
    }

    return server;
  }

  class Listener extends Thread implements ListenerTestIF {
    final Log LOG = LogFactory.getLog(Listener.class);

    private boolean shutdown = false;
    Selector acceptorSelector;
    ServerSocketChannel ssc;

    WorkerPool workerPool;

    int pipeNo = 0;
    int port;
    int pipeTimeout;
    String logPath;
    ListenerTestIF testProxy;

    Listener(NConfiguration conf, String logPath) throws IOException {
      this.port = conf.getInt("neptune.changelog.server.port", 57001) + ChangeLogServer.PORT_DIFF;
      this.pipeTimeout = conf.getInt("neptune.changelog.server.pipe.timeout", 30);
      this.logPath = logPath;

      workerPool = new WorkerPool(conf.getInt("changeLogServer.worker.count", numProcessors), port);

      acceptorSelector = Selector.open();
      ssc = ServerSocketChannel.open();

      ssc.socket().bind(new InetSocketAddress(port));
      ssc.socket().setReuseAddress(true);
      ssc.configureBlocking(false);
      LOG.info("Pipe port : " + port + ", pipe timeout : " + pipeTimeout + " sec, log path : "
          + logPath);

      testProxy = this;
      if (ChangeLogServer.IS_TEST_MODE) {
        testProxy = FaultInjectionProxy.wrap(this, ListenerTestIF.class);
      }
    }

    public int getPipePort() {
      return port;
    }

    public void stopListening() {
      LOG.info("stop listening");
      shutdown = true;
      acceptorSelector.wakeup();
      workerPool.shutdown();
    }

    public void run() {
      this.setName("listener thread [" + port + "]");

      try {
        ssc.register(acceptorSelector, SelectionKey.OP_ACCEPT);
      } catch (ClosedChannelException e) {
        e.printStackTrace();
        return;
      }

      LOG.info("Listening thread starts");

      while (!shutdown) {
        int numReady = 0;

        try {
          numReady = acceptorSelector.select();

          if (numReady > 0) {
            dispatchSelectedKeys();
          }
        } catch (IOException e) {
          LOG.warn("Exception in accepting new pipe", e);
          break;
        }
      }

      try {
        ssc.close();
        acceptorSelector.close();
      } catch (IOException e) {
        LOG.warn("closing server socket channel is fail", e);
      }

      LOG.info("Listening thread is done");
    }

    private void dispatchSelectedKeys() {
      Iterator<SelectionKey> iter = acceptorSelector.selectedKeys().iterator();

      while (iter.hasNext()) {
        SelectionKey key = iter.next();
        iter.remove();

        if (key.isValid()) {
          if (key.isAcceptable()) {
            testProxy.handleNewConnection(key);
          }
        }
      }
    }

    public void handleNewConnection(SelectionKey key) {
      SocketChannel channel = null;

      try {
        channel = ssc.accept();
      } catch (IOException e) {
        LOG.warn("error in accepting new connection due to", e);
        return;
      }

      Pipe pipe = null;

      try {
        pipe = new Pipe(channel, pipeTimeout);
        pipe.setEventHandler(makePipeEventHandler());
        pipe.setListenPortNumber(port);
        pipe.setLogPath(logPath);

        workerPool.getWorker().append(pipe);
      } catch (IOException e) {
        LOG.warn("error in creating pipe due to ", e);

        try {
          channel.close();
        } catch (IOException e1) {
          LOG.warn("closing channel is fail", e1);
        }

        return;
      }
    }

    private PipeEventHandler makePipeEventHandler() {
      return new PipeEventHandler() {
        public void pipeEstablished(String tabletName, String pipeKey, Pipe pipe)
            throws IOException {
          synchronized (tabletPipeMap) {
            HashMap<String, Pipe> pipeMap = tabletPipeMap.get(tabletName);
            if (pipeMap == null) {
              pipeMap = new HashMap<String, Pipe>();
              tabletPipeMap.put(tabletName, pipeMap);
            }
            pipeMap.put(pipeKey, pipe);
          }
          LOG.debug("Pipe key[" + pipeKey + "] is registered");

        }

        public void pipeClosed(String tabletName, String pipeKey) {
          synchronized (tabletPipeMap) {
            HashMap<String, Pipe> pipeMap = tabletPipeMap.get(tabletName);
            if (pipeMap != null) {
              pipeMap.remove(pipeKey);
              LOG.debug("Pipe key[" + pipeKey + "] is removed");
              if (pipeMap.isEmpty()) {
                tabletPipeMap.remove(tabletName);
                LOG.debug("tablet [" + tabletName + "] is removed");
              }
            }
          }
        }
      };
    }
  }

  // // RPC METHODS ////

  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
    return 1l;
  }

  public String fini(String tabletName, int bytes) {
    /*
     * try { UniversalFaultMap.instance().playFault(113); } catch (IOException
     * e1) { e1.printStackTrace(); }
     * 
     * try { //long bfGet = System.nanoTime(); Pipe pipe =
     * getAndTouchPipe(tabletName); if (pipe == null) {
     * LOG.warn("Pipe has been already closed"); return
     * "Pipe has been already closed"; } else if (!pipe.isActivated()) {
     * LOG.warn("Pipe is not activated"); return "Pipe is not activated"; }
     * 
     * //long afGet = System.nanoTime();
     * 
     * pipe.fini(bytes);
     * 
     * //long afFini = System.nanoTime();
     * 
     * //LOG.debug("get : " + calc(afGet, bfGet) + ", fini : " + calc(afFini,
     * afGet)); } catch (Exception e) {
     * LOG.warn("Abort finishing writing... tablet : " + tabletName +
     * ", bytes : " + bytes, e); return "ABORT"; }
     * 
     * try { UniversalFaultMap.instance().playFault(115); } catch (IOException
     * e1) { e1.printStackTrace(); }
     */
    return "OK";
  }

  double calc(long after, long before) {
    return (after - before) / 1000000.0;
  }

  public String rollback(String tabletName, String position) {
    LOG.info("rollback is called! with tabletName : " + tabletName);
    RandomAccessFile file = null;
    try {
      File logFile = new File(changeLogStorePath + File.separator + tabletName + File.separator
          + Constants.PIPE_CL_FILE_NAME + listener.port);

      file = new RandomAccessFile(logFile, "rws");
      file.getChannel().truncate(Long.parseLong(position));
    } catch (IOException e) {
      LOG.warn("rollbacking file is fail tablet [" + tabletName + "], pos : " + position, e);
      return "ABORT";
    } finally {
      if (file != null) {
        try {
          file.close();
        } catch (IOException e) {
        }
      }
    }

    return "OK";
  }

  public String backupLogFile(String tabletName) {
    LOG.info("backupLogFile enters tablet[" + tabletName + "]");
    try {
      ChangeLogFileChannel.backup(tabletName, changeLogStorePath, Constants.PIPE_CL_FILE_NAME + listener.port);
    } catch(IOException e) {
      LOG.warn("fail to backupLogFile of tablet [" + tabletName + "]");
      return "NOT OK";
    } 

    LOG.info("backupLogFile exits tablet[" + tabletName + "]");

    return "OK";
  }

  public void restoreLogFile(String tabletName) {
    ChangeLogFileChannel channel = null;
    try {
      channel = ChangeLogFileChannel.createChannel(tabletName, changeLogStorePath, 
          Constants.PIPE_CL_FILE_NAME + listener.port);
      channel.restoreLastLog();
    } catch(IOException e) {
      LOG.warn("fail to restoreLogFile of tablet [" + tabletName + "]");
    } finally {
      if (channel != null) {
        try {
          ChangeLogFileChannel.closeChannel(channel);
        } catch (IOException e) {
          LOG.warn("fail to close change log file channel due to " + e);
        }
      }
    }

  }

  // by sangchul : 같은 VM에서 실행되는 ChangeLogServer들의 동기화를 위해
  static final ReentrantLock formatLock = new ReentrantLock();

  public void format() {
    final File file = new File(changeLogStorePath);

    formatLock.lock();

    final File tempFile = new File(makeTempLogStorePath());

    try {
      if (file.exists() == false) {
        return;
      }

      if (file.renameTo(tempFile) == false) {
        LOG.warn("Fail to format!");
        return;
      }
    } finally {
      formatLock.unlock();
    }

    createCleanThread(tempFile).start();
    LOG.info("Starting file format thread");
  }

  private Thread createCleanThread(final File tempFile) {
    return new Thread() {
      public void run() {
        clean(tempFile);
        LOG.info("Finishing file format thread");
      }

      private void clean(File file) {
        if (file.isDirectory()) {
          for (File childrenFile : file.listFiles()) {
            clean(childrenFile);
          }
        }

        if (file.delete() == false) {
          LOG.warn("Deleting " + file.getAbsolutePath() + " is fail");
        }
      }
    };
  }

  private String makeTempLogStorePath() {
    String result;
    if (changeLogStorePath.charAt(changeLogStorePath.length() - 1) == '\\'
        || changeLogStorePath.charAt(changeLogStorePath.length() - 1) == '/') {
      result = changeLogStorePath.substring(0, changeLogStorePath.length() - 1);
    } else {
      result = changeLogStorePath;
    }

    return result + "_" + System.currentTimeMillis();
  }

  public String removeAllLogs(final String tabletName) throws IOException {
    LOG.debug("removeAllLogs is called! with tabletName : " + tabletName);

    final File tabletDir = new File(changeLogStorePath + File.separator + tabletName);
    if (tabletDir.exists()) {
      Thread removeAllLogsThread = new Thread() {
        public void run() {
          try {
            LOG.info("removeAll tabletName[" + tabletName + "] enters");
            for (File file : tabletDir.listFiles()) {
              if (!file.delete()) {
                LOG.warn("fail to delete log file [" + file.getAbsolutePath() + "]");
              }
            }

            if (!tabletDir.delete()) {
              LOG.warn("fail to delete log file [" + tabletDir.getAbsolutePath() + "]");
            }
		  } catch(Exception e) {
            LOG.error("fail to remove all logs tablet[" + tabletName + "]", e);
          } finally {
            LOG.info("removeAll tabletName[" + tabletName + "] exits");
          }
        }
      };
 
      removeAllLogsThread.start();
    }

    return "OK";
  }

  public void removeBackupLogs(String tabletName) throws IOException {
    ChangeLogFileChannel channel = null;
    try {
      channel = ChangeLogFileChannel.createChannel(tabletName, changeLogStorePath, 
          Constants.PIPE_CL_FILE_NAME + listener.port);
      channel.removeBackupFiles();
    } catch(IOException e) {
      LOG.warn("fail to removeBackupFiles of tablet [" + tabletName + "]");
    } finally {
      if (channel != null) {
        try {
          ChangeLogFileChannel.closeChannel(channel);
        } catch (IOException e) {
          LOG.warn("fail to close change log file channel due to " + e);
        }
      }
    }
  }

  private File getLogFile(String tabletName) {
    File parentDir = new File(changeLogStorePath + File.separator + tabletName + File.separator);

    return new File(parentDir, Constants.PIPE_CL_FILE_NAME + listener.port);
  }

  public long getLogFileSize(String tabletName) {
    LOG.debug("getLogFileSize is called! with tabletName : " + tabletName);

    File log = getLogFile(tabletName);
    if (log.exists() == false) {
      LOG.warn("getting file size of tablet[" + tabletName + "] is fail. file does not exist");
    }

    return log.length();
  }

  public void closePipe(String tabletName, String pipeKey) {
    LOG.debug("close is called! with tabletName : " + tabletName + ", pipeKey : " + pipeKey);
    long before = System.currentTimeMillis();
    synchronized (tabletPipeMap) {
      HashMap<String, Pipe> pipeMap = tabletPipeMap.get(tabletName);
      if (pipeMap != null) {
        Pipe pipe = pipeMap.get(pipeKey);
        if (pipe != null) {
          pipe.triggerClose();
        }
      }
    }

    long elapsed = System.currentTimeMillis() - before;
    if (elapsed > 2000) {
      LOG.warn("It takes too long to trigger pipe closing. key : [" + pipeKey + "] time : "
          + elapsed + "ms");
    }
  }

  public void closeAll(String tabletName) {
    LOG.debug("close is called! with tabletName : " + tabletName);
    long before = System.currentTimeMillis();
    synchronized (tabletPipeMap) {
      for (Pipe pipe : tabletPipeMap.get(tabletName).values()) {
        LOG.debug("trigger closing pipe#" + pipe.getPipeKey());
        pipe.triggerClose();
      }
    }

    long elapsed = System.currentTimeMillis() - before;
    if (elapsed > 2000) {
      LOG.warn("It takes too long to trigger all the pipes closing. time : " + elapsed + "ms");
    }
  }

  public void stopChangeLogServer() {
    try {
      this.stop();
    } catch (java.lang.reflect.UndeclaredThrowableException e) {
    }

    LOG.info("Change log server is stopped");
    System.exit(-1);
  }

  public int readAllLogs(String tabletName) {
    LOG.debug("readAllLogs is called with tabletName : " + tabletName);
    File file = new File(changeLogStorePath + tabletName);

    if (file.exists() == false || file.list().length == 0) {
      LOG.warn("A tablet directory does not exist");
      return -1;
    }

    return startFileTransferChannel(file.listFiles());
  }

  public int readLastLog(String tabletName) {
    LOG.debug("readLastLog is called with tabletName : " + tabletName);
    File file = new File(changeLogStorePath + tabletName);

    if (file.exists() == false) {
      LOG.warn("A tablet directory does not exist");
      return -1;
    }

    File[] fileList = file.listFiles();
    if (fileList.length == 0) {
      LOG.warn("file does not exist");
      return -1;
    }

    File lastFile = null;
    long lastModified = 0l;

    for (int i = 0; i < fileList.length; i++) {
      if (fileList[i].lastModified() > lastModified) {
        lastFile = fileList[i];
        lastModified = lastFile.lastModified();
      }
    }

    if (lastFile.length() == 0) {
      LOG.warn("the length of the file is zero");
      return -1;
    }

    return startFileTransferChannel(new File[] { lastFile });
  }

  public String tabletExists(String tabletName) {
    LOG.debug("tabletExists is called with tabletName : " + tabletName);
    File file = new File(changeLogStorePath + tabletName);

    if (file.exists()) {
      LOG.debug("tablet exists : " + String.valueOf(file.list().length > 0));
      return String.valueOf(file.list().length > 0);
    }

    LOG.info("Tablet : " + tabletName + " does not exist");
    return "false";
  }

  private int startFileTransferChannel(File[] fileList) {
    int port = -1;
    try {
      ServerSocketChannel ssc = ServerSocketChannel.open();
      ssc.configureBlocking(false);
      ServerSocket serverSocket = ssc.socket();
      serverSocket.bind(null);
      port = serverSocket.getLocalPort();

      new FileTransferThread(ssc, fileList).start();
      LOG.info("File transfer thread is started and read method is done");
    } catch (IOException e) {
      LOG.warn("starting file transfer is fail", e);
    }

    return port;
  }

  private Pipe getAPipeOf(String tabletName) throws IOException {
    synchronized (tabletPipeMap) {
      HashMap<String, Pipe> pipeMap = tabletPipeMap.get(tabletName);

      if (pipeMap == null) {
        throw new IOException("Tablet[" + tabletName + "] is not served in this changelogserver");
      }

      Iterator<Pipe> iter = pipeMap.values().iterator();
      if (iter.hasNext()) {
        Pipe pipe = iter.next();
        return pipe;
      }

      throw new IOException("Tablet[" + tabletName + "] is not served in this changelogserver");
    }
  }

  public long getChecksum(String tabletName) throws IOException {
    File file = new File(changeLogStorePath + tabletName + File.separator
        + Constants.PIPE_CL_FILE_NAME + (rpcPort + PORT_DIFF));
    long fileLength = file.length();

    CheckedInputStream cksumIn = new CheckedInputStream(new FileInputStream(file), new CRC32());
    BufferedInputStream in = new BufferedInputStream(cksumIn, 8192);

    for (long i = 0; i < fileLength; i++) {
      in.read();
    }

    return cksumIn.getChecksum().getValue();
  }

  public void test() {
  }
}
